package com.gagakuai.bpm.domian.events;

import com.alibaba.fastjson.JSON;

import com.gagakuai.bpm.common.component.PublisherContext;
import com.gagakuai.bpm.common.constants.BpmMetaConstants;
import com.gagakuai.bpm.common.constants.RabbitMqConstants;
import com.gagakuai.bpm.dao.mapper.BpmDeadMessageLogMapper;
import com.gagakuai.bpm.domian.entity.BpmDeadMessageLog;
import com.gagakuai.bpm.domian.wapper.DeadMessageWrapper;
import com.gagakuai.bpm.mqlistener.camunda.BpmMessageBusnessListener;
import com.gagakuai.bpm.utils.ObjectUtils;
import com.gagakuai.bpm.utils.RedisUtil;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.ApplicationListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import java.util.Date;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicBoolean;

/*
 * @program: bpm_server
 * @description: 死信队列监听器:处理死信
 * @author: houhong
 * @create: 2022-07-11 09:40
 */
@Slf4j
@Component
public class DeadMessageListener implements ApplicationListener<DeadMessageEvent> {

    @Autowired
    @Qualifier(value = BpmMetaConstants.DEAD_LOCAL_QUEUE_NAME)
    private Queue<DeadMessageWrapper> deadMessageEventQueue;

    @Autowired
    private BpmDeadMessageLogMapper bpmDeadMessageLogMapper;

    @Autowired
    private BpmMessageBusnessListener bpmMessageBusnessListener;

    @Autowired
    private RedisUtil redisUtil;

    @Autowired
    private PublisherContext publisherContext;

    //开启重试开关
    private static AtomicBoolean RETRY_SWITCH = new AtomicBoolean(false);


    @Async(value = "deadQueueThreadPool")
    @Override
    @Transactional(propagation = Propagation.REQUIRED, rollbackFor = Exception.class)
    public void onApplicationEvent(DeadMessageEvent deadMessageEvent) {

        log.info("======开始处理死信消息======");
        DeadMessageWrapper deadMessageWrapper = deadMessageEventQueue.poll();
        Message message = deadMessageWrapper.getMessage();
        String messageType = deadMessageWrapper.getType();

        //channel 应该换成死信的
        Channel channel = deadMessageWrapper.getChannel();


        String correlationId = deadMessageWrapper.getCorrelationId();


        log.info("#当前处理的数据队列类型,{},当前消息id:{}#", messageType, correlationId);

        //如果在这里用系统尝试做一次补偿次数,使用correlationId
        BpmDeadMessageLog bpmDeadMessageLogDb = bpmDeadMessageLogMapper.selectByMsgId(correlationId);

        if (ObjectUtils.isNull(bpmDeadMessageLogDb)) {
            //直接进行入库操作
            BpmDeadMessageLog bpmDeadMessageLog = new BpmDeadMessageLog();
            bpmDeadMessageLog.setBpmCreateTime(new Date());
            bpmDeadMessageLog.setBpmRetryCount(0);
            bpmDeadMessageLog.setBpmDeadMessageStatus(0);
            bpmDeadMessageLog.setBpmMsgId(correlationId);
            //重试消息类别
            bpmDeadMessageLog.setBpmMessageType(messageType);

            String messageJsonStr = JSON.toJSONString(message);
            bpmDeadMessageLog.setBpmMessageInfo(messageJsonStr);
            //用户当前ip: // TODO: 2022/7/13  -- 做saas化
            bpmDeadMessageLog.setBpmTantIp(null);


            /*
             *
             *   使用redis做中间状态扭转 -- 问题：如果网络崩溃，依靠redis做的所有中间状态全部问题，
             *      怎么做熔断/保护？
             */
            String retryKey = BpmMetaConstants.BPM_DEAD_MSG_SYSTEM_RETRY
                    + BpmMetaConstants.REDIS_FILE_KEY_SEPTOR
                    + correlationId;
            //不存在key ---> 表示未做系统重试，重试一次,在代理类里面成功后，去要删除

            Boolean systemRetryFlag = false;
            if (!redisUtil.hasKey(retryKey)) {
                //value表示重试
                redisUtil.set(retryKey, correlationId + ":" + 1);
                //系统重试 --> 如果出问题的化，就直接再重试一次就好了，就不这么写代码,这里没有必要，因为chaennel 已经被关闭了。
                //   systemRetryFlag = systemRetry(channel, message, messageType);
            }

            //失败
            if (!systemRetryFlag) {
                bpmDeadMessageLogMapper.insertSelective(bpmDeadMessageLog);
                /*
                 *
                 *  todo BPM系统在这里发布一个消息出去。业务系统监听这个消息队列,做业务回调，在监听发布出去后，
                 *       业务系统根据工作流类型,和业务表类型，做回滚补偿,怎么saas 化？怎么和业务挂钩？，怎么和业务解耦？
                 *       //// TODO: 2022/7/13
                 *                  将一些业务系统需要开发的东西，但是又是一些通用的东西全部用springboot 的自动化配置封装成starter
                 *                  然后将通用功能，比如业务系统等回调方法。
                 *                  定义注解/spi机制 等 让业务系统接入
                 *                  比如这里，业务系统需要回滚状态的时候，在@Bpmrollback(table ='xxxx',isEnable = true,....)
                 *                  就行了。对于其他的东西也是一样。  全部进行抽象出来。第二期做，现在先保证上线。
                 */
                BusinessSysCallBackEvent.BusinessSysCallBackEventData eventData = new BusinessSysCallBackEvent.BusinessSysCallBackEventData();
                eventData.setBpmDeadMessageLog(bpmDeadMessageLog);
                //拼接数据
                BusinessSysCallBackEvent sysCallBackEvent = new BusinessSysCallBackEvent(eventData);
                publisherContext.publish(sysCallBackEvent);
            }

            try {
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                //删除
                redisUtil.del(retryKey);
            } catch (Exception e) {
                e.printStackTrace();
                log.error("#手动确认死信队列异常,错误西信息如下:{}#", e.getMessage());
            }
        } else {
            log.info("#幂等处理#");
        }
        log.info("======死信消息处理成功======");
    }

    /*
     * 获取CorrelationId
     *
     * @param message
     * @return
     */
    private String getCorrelationId(Message message) {

        String correlationId = null;

        MessageProperties properties = message.getMessageProperties();
        Map<String, Object> headers = properties.getHeaders();
        for (Map.Entry entry : headers.entrySet()) {
            String key = (String) entry.getKey();
            String value = (String) entry.getValue();
            if (key.equals("spring_returned_message_correlation")) {
                correlationId = value;
            }
        }

        return correlationId;
    }

    public static void main(String[] args) {
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setHeader("name", "houhong");
        messageProperties.setHeader("age", 26);
        Message message = new Message("houh".getBytes(), messageProperties);

        String s = JSON.toJSONString(message);
        System.out.println(s);
    }

    /*
     * @return
     * @Author houhong
     * @Description //TODO 线程重试
     * @Date 5:09 下午 2022/7/13
     */
    public void threadRetryTask() {

        ExecutorService retrySystem = Executors.newSingleThreadExecutor(new ThreadFactory() {

            @Override
            public Thread newThread(Runnable r) {

                Thread thread = new Thread(r);
                thread.setName("bpm-hx-system-retry");

                return null;
            }
        });
        if (RETRY_SWITCH.compareAndSet(false, true)) {

        }
    }


    /*
     * @param message     信息
     * @param messageType 类型
     * @param channel     信道
     * @return {@link Boolean}
     * @Author houhong
     * @Description //TODO  在插入表之前做一次系统重试,以后业务多了以后，在这里要扩展
     * @Date 4:50 下午 2022/7/13
     */
    public Boolean systemRetry(Channel channel, Message message, String messageType) {

        Boolean systemRetry = true;
        try {
            switch (messageType) {
                //撤销流程
                case RabbitMqConstants.BPM_WITH_DRAW_QUEUE_NAME:
                    bpmMessageBusnessListener.consumerWithDrawmeaasge(message, channel);
                    break;
                //退回上一步
                case RabbitMqConstants.BPM_WITH_DRAW_FIRST_STEP_QUEUE_NAME:
                    bpmMessageBusnessListener.consumerWithDrawFirstStepmeaasge(message, channel);
                    break;
                //退回重填
                case RabbitMqConstants.BPM_WITH_DRAW_LAST_STEP_QUEUE_NAME:
                    bpmMessageBusnessListener.consumerWithDrawLastStepmeaasge(message, channel);
                    break;

                //取回
                case RabbitMqConstants.BPM_WITH_RETURN_QUEUE_NAME:
                    bpmMessageBusnessListener.consumerWithReturnMeaasge(message, channel);
                    break;
                default:
                    break;
            }
        } catch (Exception e) {
            log.error("#重试一次错误:{}#", e.getMessage());
            systemRetry = false;
        }

        return systemRetry;
    }

}